package com.huatai.bi.service.impl;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.huatai.bi.constant.RabbitmqConstant;
import com.huatai.bi.dto.MqttClientDto;
import com.huatai.bi.entity.DirectoryEntity;
import com.huatai.bi.entity.MqMqttClientEntity;
import com.huatai.bi.mapper.MqMqttClientMapper;
import com.huatai.bi.mqtt.MqttClientCallback;
import com.huatai.bi.mqtt.MqttClientConnect;
import com.huatai.bi.service.DirectoryService;
import com.huatai.bi.service.MqMqttClientService;
import com.huatai.bi.utils.PaginationUtils;
import com.huatai.common.api.R;
import com.huatai.common.error.exception.ServiceException;
import jodd.util.StringUtil;
import lombok.extern.slf4j.Slf4j;
//import org.bson.Document;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.mongodb.core.MongoTemplate;
//import org.springframework.data.mongodb.core.query.Criteria;
//import org.springframework.data.mongodb.core.query.Query;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @author songxiude
 * @description 针对表【mq_mqtt_client(mqtt链接信息表)】的数据库操作Service实现
 * @createDate 2024-07-15 14:36:47
 */
@Service
@Slf4j
public class MqMqttClientServiceImpl extends ServiceImpl<MqMqttClientMapper, MqMqttClientEntity>
	implements MqMqttClientService {

	@Autowired
	@Lazy
	private DirectoryService directoryService;

	@Autowired
	private RabbitTemplate rabbitTemplate;

//	@Autowired
//	private MongoTemplate mongoTemplate;

	/**
	 * 添加修改mqtt链接
	 *
	 * @param mqMqttClient
	 * @return
	 */
	@Transactional(rollbackFor = Exception.class)
	public R mqttSave(MqMqttClientEntity mqMqttClient) {
		//判断链接是否已存在
//		long count = this.count(Wrappers.<MqMqttClientEntity>lambdaQuery().eq(MqMqttClientEntity::getClientId, mqMqttClient.getClientId()));
//		if (ObjectUtils.isEmpty(mqMqttClient.getId()) && count > 0) {
//			throw new ServiceException("该链接已存在不可重复创建");
//		}
		if(mqMqttClient.getIsParent() != null && mqMqttClient.getIsParent().intValue() == 1 && StringUtils.isEmpty(mqMqttClient.getMarkInfo())) {
			throw new ServiceException("父子终端层级标识不能为空");
		}
		//判断链接是否已经存在
		long count = this.count(Wrappers.<MqMqttClientEntity>lambdaQuery().eq(MqMqttClientEntity::getServerHost, mqMqttClient.getServerHost())
			.eq(MqMqttClientEntity::getServerPort, mqMqttClient.getServerPort()).like(MqMqttClientEntity::getClientTopic, mqMqttClient.getClientTopic()));
		if (count > 0) {
			throw new ServiceException("该链接已存在不可重复创建");
		}
		//判断topic是否存在
//		long tcount = this.count(Wrappers.<MqMqttClientEntity>lambdaQuery().like(MqMqttClientEntity::getClientTopic, mqMqttClient.getClientTopic()));
//		if (tcount > 0) {
//			throw new ServiceException("所订阅的topic已经存在");
//		}
		//判断数据集是不是目录
		DirectoryEntity directoryEntity = directoryService.getOne(Wrappers.<DirectoryEntity>lambdaQuery().eq(DirectoryEntity::getDtId, mqMqttClient.getDtId()));
		if (ObjectUtils.isEmpty(directoryEntity) || directoryEntity.getIsDir().intValue() != 0) {
			throw new ServiceException("数据集不是目录不能添加新的链接");
		}
		//组装链接地址前缀
		if (mqMqttClient.getClientType().intValue() == 1) {
			mqMqttClient.setMqttPrefix("ssl://");
		} else if (mqMqttClient.getClientType().intValue() == 2){
			mqMqttClient.setMqttPrefix("tcp://");
		}else if (mqMqttClient.getClientType().intValue() == 3){
			mqMqttClient.setMqttPrefix("wss://");
		}
		//如果不传客户端链接id自动生成
		if (StringUtil.isEmpty(mqMqttClient.getClientId())) {
			mqMqttClient.setClientId(IdWorker.get32UUID());
		}
		//链接mqtt，默认未连接
		mqMqttClient.setStatus(0);
//		CompletableFuture.runAsync(() -> {
//			boolean mqB = this.initMqttServer(mqMqttClient);
//			if (!mqB) {
//				mqMqttClient.setStatus(3);
//			} else {
//				mqMqttClient.setStatus(1);
//			}
//			save(mqMqttClient);
//		});

		mqMqttClient.setEnableFlag(true);
		//改造为最少连接客户端触发连接
		save(mqMqttClient);
//		if (save) {
//			JSONObject jsonObject = new JSONObject();
//			jsonObject.put("type", 1);
//			jsonObject.put("data", mqMqttClient);
//			rabbitTemplate.convertAndSend(RabbitmqConstant.RABBIT_MQTT_FANOUT, "", jsonObject.toJSONString());
//			return save;
//		}
		return R.data(mqMqttClient);
	}

	/**
	 * 修改mqtt链接信息
	 *
	 * @param mqMqttClient
	 * @return
	 */
	@Transactional(rollbackFor = Exception.class)
	public boolean mqttUpdate(MqMqttClientEntity mqMqttClient) {
		//判断topic是否存在
		long tcount = this.count(Wrappers.<MqMqttClientEntity>lambdaQuery()
			.like(MqMqttClientEntity::getClientTopic, mqMqttClient.getClientTopic())
			.ne(MqMqttClientEntity::getId, mqMqttClient.getId()));
		if (tcount > 0) {
			throw new ServiceException("所订阅的topic已经存在");
		}
		//先停掉运行中的链接
		ConcurrentHashMap<String, MqttClientConnect> mqttClients = MqttClientConnect.mqttClients;
		MqttClientConnect clientConnect = mqttClients.get(mqMqttClient.getClientId());
		if (ObjectUtil.isNotEmpty(clientConnect)) {
			try {
				clientConnect.close();
				mqttClients.remove(mqMqttClient.getClientId());
			} catch (MqttException e) {
				log.error("关闭链接失败");
				throw new ServiceException("关闭链接失败");
			}
		}

		//组装链接地址前缀
		if (mqMqttClient.getClientType().intValue() == 1) {
			mqMqttClient.setMqttPrefix("ssl://");
		} else if (mqMqttClient.getClientType().intValue() == 2) {
			mqMqttClient.setMqttPrefix("tcp://");
		} else if(mqMqttClient.getClientType().intValue() == 3){
			mqMqttClient.setMqttPrefix("wss://");
		}
		//链接mqtt
		mqMqttClient.setStatus(2);
		CompletableFuture.runAsync(() -> {
			boolean mqB = this.initMqttServer(mqMqttClient);
			if (!mqB) {
				mqMqttClient.setStatus(3);
			} else {
				mqMqttClient.setStatus(1);
			}
			updateById(mqMqttClient);
		});
		updateById(mqMqttClient);
//		if(update) {
//			JSONObject jsonObject = new JSONObject();
//			jsonObject.put("type", 2);
//			jsonObject.put("data", mqMqttClient);
//			rabbitTemplate.convertAndSend(RabbitmqConstant.RABBIT_MQTT_FANOUT, "", jsonObject.toJSONString());
//			return update;
//		}
		return true;
	}

	/**
	 * 获取mqtt链接信息
	 *
	 * @param mqttClientDto
	 * @return
	 */
	public IPage<MqMqttClientEntity> mqttList(MqttClientDto mqttClientDto) {
		Page<MqMqttClientEntity> page = PaginationUtils.<MqMqttClientEntity>getPage(mqttClientDto.getPage(), mqttClientDto.getSize());
		LambdaQueryWrapper<MqMqttClientEntity> wrapper = Wrappers.<MqMqttClientEntity>lambdaQuery();
		//排除部分敏感字段
		//wrapper.select(MqMqttClientEntity.class, client -> !client.getColumn().equals("password"));
		//wrapper.select(MqMqttClientEntity::getId,MqMqttClientEntity::getMqttName);
		wrapper.like(StringUtil.isNotEmpty(mqttClientDto.getMqttName()), MqMqttClientEntity::getMqttName, mqttClientDto.getMqttName());
		wrapper.orderByDesc(MqMqttClientEntity::getCreateTime);
		return page(page, wrapper);
	}

	/**
	 * 手动启动mqtt
	 *
	 * @param id
	 * @return
	 */
	public boolean startMqtt(Long id) {
		MqMqttClientEntity clientEntity = getById(id);
		if (ObjectUtils.isEmpty(clientEntity)) {
			throw new ServiceException("不存在的实例");
		}
		if (clientEntity.getStatus().intValue() == 1 || clientEntity.getStatus().intValue() == 2) {
			throw new ServiceException("已连接或连接中，请勿重复操作");
		}
		if(clientEntity.getStatus().intValue() == 3){
			throw new ServiceException("连接信息异常，请核查");
		}
		clientEntity.setStatus(2);
		clientEntity.setEnableFlag(true);
		boolean update = updateById(clientEntity);
		CompletableFuture.runAsync(() -> {
			boolean mqB = this.initMqttServer(clientEntity);
			if (!mqB) {
				clientEntity.setStatus(3);
			} else {
				clientEntity.setStatus(1);
			}
			updateById(clientEntity);
		});
//		if(update){
//			JSONObject jsonObject = new JSONObject();
//			jsonObject.put("type", 4);
//			jsonObject.put("data", clientEntity);
//			rabbitTemplate.convertAndSend(RabbitmqConstant.RABBIT_MQTT_FANOUT, "", jsonObject.toJSONString());
//			return update;
//		}
		return update;
	}

	/**
	 * 删除mqtt链接
	 *
	 * @param clientId
	 * @return
	 */
	@Transactional(rollbackFor = Exception.class)
	public boolean removeMqtt(String clientId) {
		try {
			ConcurrentHashMap<String, MqttClientConnect> mqttClients = MqttClientConnect.mqttClients;
			MqttClientConnect mqttClientConnect = mqttClients.get(clientId);
			if (ObjectUtil.isNotEmpty(mqttClientConnect)) {
				mqttClientConnect.close();
			}
			mqttClients.remove(clientId);
		} catch (MqttException e) {
			e.printStackTrace();
			log.error("断开连接失败,id--->" + clientId);
			//return false;
		}
		MqMqttClientEntity mqttClientEntity = getOne(Wrappers.<MqMqttClientEntity>lambdaQuery().eq(MqMqttClientEntity::getClientId, clientId));
		if (ObjectUtil.isEmpty(mqttClientEntity)) {
			throw new ServiceException("非法删除");
		}
		long count = directoryService.count(Wrappers.<DirectoryEntity>lambdaQuery().eq(DirectoryEntity::getDtId, mqttClientEntity.getDataSetId()));
		if (count > 0) {
			throw new ServiceException("已存在数据集，请删除数据集");
		}
		boolean remove = remove(Wrappers.<MqMqttClientEntity>lambdaQuery().eq(MqMqttClientEntity::getClientId, clientId));
//		if(remove){
//			JSONObject jsonObject = new JSONObject();
//			jsonObject.put("type", 5);
//			jsonObject.put("data", mqttClientEntity);
//			rabbitTemplate.convertAndSend(RabbitmqConstant.RABBIT_MQTT_FANOUT, "", jsonObject.toJSONString());
//			return remove;
//		}
		return remove;
	}

	/**
	 * 初始化链接mqtt
	 *
	 * @param mqMqttClient
	 */
	public boolean initMqttServer(MqMqttClientEntity mqMqttClient) {
		try {
			MqttClientConnect mqttClientConnect = new MqttClientConnect();
			mqttClientConnect.setMqttClientId(mqMqttClient.getClientId());
			mqttClientConnect.setMqttClient(mqMqttClient, new MqttClientCallback(mqMqttClient.getClientId()));
			mqttClientConnect.sub(StringUtils.commaDelimitedListToStringArray(mqMqttClient.getClientTopic()));
			MqttClientConnect.mqttClients.put(mqMqttClient.getClientId(), mqttClientConnect);
			log.info("mqtt初始化链接成功,host--->" + mqMqttClient.getServerHost());
			return true;
		} catch (Exception e) {
			e.printStackTrace();
			log.error("新增连接失败,host--->" + mqMqttClient.getServerHost());
			return false;
		}
	}

	/**
	 * 停止Mqtt
	 *
	 * @param id
	 * @return
	 */
	public boolean stopMqtt(Long id) {
		MqMqttClientEntity clientEntity = getById(id);
		if (ObjectUtils.isEmpty(clientEntity)) {
			throw new ServiceException("不存在的链接实例");
		}
		//先停止mqtt链接
		ConcurrentHashMap<String, MqttClientConnect> mqttClients = MqttClientConnect.mqttClients;
		MqttClientConnect mqttClientConnect = mqttClients.get(clientEntity.getClientId());
		if (!ObjectUtils.isEmpty(mqttClientConnect)) {
			try {
				mqttClientConnect.close();
				mqttClients.remove(clientEntity.getClientId());
			} catch (MqttException e) {
				throw new ServiceException("停止mqtt链接失败");
			}
		}
		//更新状态
		clientEntity.setStatus(0);
		clientEntity.setEnableFlag(false);
		boolean update = updateById(clientEntity);
//		if(update) {
//			JSONObject jsonObject = new JSONObject();
//			jsonObject.put("type", 3);
//			jsonObject.put("data", clientEntity);
//			rabbitTemplate.convertAndSend(RabbitmqConstant.RABBIT_MQTT_FANOUT, "", jsonObject.toJSONString());
//			return update;
//		}
		return update;
	}

	/**
	 * 获取mqtt连接最小的实例
	 * @param ips
	 * @return
	 */
	public JSONObject getMinClient(List<String> ips) {
		return baseMapper.getMinClient(ips);
	}

	@Override
	public R getMqttTree(MqMqttClientEntity dto) {
		List<MqMqttClientEntity> dataList = this.baseMapper.selectList(
			new LambdaQueryWrapper<MqMqttClientEntity>()
				.like(StringUtil.isNotBlank(dto.getMqttName()),MqMqttClientEntity::getMqttName, dto.getMqttName())
		);

		List<JSONObject> resultList = new ArrayList<>();
		if(dataList.size() == 0){
			return R.data(resultList);
		}

		List<DirectoryEntity> entities = directoryService.getBaseMapper().selectList(
			new LambdaQueryWrapper<DirectoryEntity>()
				.eq(DirectoryEntity::getIsDeleted,0)
				.eq(DirectoryEntity::getDataType,1)
		);
		if(entities.size() == 0){
			return R.data(resultList);
		}
		Map<String,List<DirectoryEntity>> entityMap = entities.stream().collect(Collectors.groupingBy(DirectoryEntity::getClientId));

		for (MqMqttClientEntity entity : dataList) {
			JSONObject jsonObject = new JSONObject();
			jsonObject.put("id", entity.getId());
			jsonObject.put("name", entity.getMqttName());
			jsonObject.put("topic", entity.getClientTopic());
			jsonObject.put("isParent", entity.getIsParent());

			List<DirectoryEntity> twoList = null;
			if(entity.getIsParent().intValue() == 1){
				twoList =  entityMap.get(entity.getClientId());
			}
			List<JSONObject> twoJsonList = new LinkedList<>();
			String[] topics = entity.getClientTopic().split(",");
			for (String topic : topics) {
				JSONObject two = new JSONObject();
				two.put("topic", topic);
				if(twoList != null){
					Map<String,List<DirectoryEntity>> twoEntityMap = twoList.stream().collect(Collectors.groupingBy(DirectoryEntity::getTopicInfo));
					two.put("child", twoEntityMap.get(topic) != null ? twoEntityMap.get(topic) : new JSONArray());
				}else{
					two.put("child", new JSONArray());
				}
				twoJsonList.add(two);
			}
			jsonObject.put("child", twoJsonList);
			resultList.add(jsonObject);
		}

		return R.data(resultList);
	}

//	@Override
//	public R getclientIdAllDate(String clientId) {
//		Query query = new Query(Criteria.where("clientId").is(clientId));
//		List<Document> documents = mongoTemplate.find(query, Document.class, "mqtt_data");
//
//		List<Map<String, Object>> resultList = new ArrayList<>();
//
//		for (Document doc : documents) {
//			Document data = (Document) doc.get("data");
//			if (data == null) continue;
//
////			Long timestamp = data.getLong("timestamp");
//
////			Document events = (Document) data.get("events");
////			if (events == null) continue;
//
////			Document defaultData = (Document) events.get("default");
////			if (defaultData == null) continue;
//
//			Map<String, Object> resultItem = new HashMap<>(data);
//			resultList.add(resultItem);
//		}
//
//		return R.data(resultList); // 假设你用的是通用响应封装类 R
//	}


}




